package com.xunk.equipment.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.xunk.common.core.constant.DnakeConstants;
import com.xunk.common.redis.service.RedisService;
import com.xunk.equipment.annotation.MqttListener;
import com.xunk.equipment.domain.EquipmentDO;
import com.xunk.equipment.mapper.EquipmentMapper;
import com.xunk.equipment.service.BaseService;
import com.xunk.equipment.service.dnake.vo.DnakeConnectMqttMessage;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.*;

/**
 * @ProjectName: xunk
 * @Package: com.xunk.equipment.listener
 * @ClassName: DnakeListener
 * @Author: wangTong
 * @Description: 狄耐克订阅topic 网关上线离线
 * @Date: 2021/6/8 10:02
 * @Version: 1.0
 */
@Slf4j
@Service
@MqttListener(topic = "connect/+/+", type = 2)
public class DnakeConnectListener implements IMqttMessageListener {

    @Resource(name = "asyncMqttTaskExecutor")
    ThreadPoolTaskExecutor threadPoolTaskExecutor;
    @Autowired
    RedisService redisService;
    @Autowired
    BaseService baseService;
    @Autowired
    EquipmentMapper equipmentMapper;
//    @Autowired
//    DnakePublishComponent dnakePublishComponent;

    @Override
    public void messageArrived(String topic, MqttMessage message) {
        threadPoolTaskExecutor.execute(() -> {
            DnakeConnectMqttMessage response = JSONObject.parseObject(new String(message.getPayload()),
                    new TypeReference<DnakeConnectMqttMessage>() {
                    });
            log.info("=========== {}", JSON.toJSONString(response));
            String[] topics = topic.split("/");
            String isConnected = topics[1];
            String userName = response.getUsername();
            if (userName.contains("&")) {
                userName = userName.substring(0, userName.indexOf("&"));
            }
            // 网关上线
            if(DnakeConstants.ONLINE.equals(isConnected)){
                // 更新数据库 网关设备上线
                equipmentMapper.updateEquipmentOnline(userName, 1L);

                // 发送mqtt信息查询网关子设备状态
//                dnakePublishComponent.simplePublish(DnakeActionEnum.READALLDEVSTATUS.getCode(),userName, DnakeConstants.ONLINE);
                // webSocket发送给APP 暂时不写

            }
            // 网关离线
            if(DnakeConstants.OFFLINE.equals(isConnected)){
                // 更新数据库 网关设备以及网关子设备均离线
                equipmentMapper.updateGatewayOnline(userName, 0L);
                // 更新redis
                // 查询网关子设备列表
                List<EquipmentDO> equipmentDOS = equipmentMapper.listGatewayEquipments(userName);
                equipmentDOS.forEach(equipmentDO -> redisInfo(equipmentDO.getDeptId(), equipmentDO.getDeviceSn(), "online", "0"));

                // webSocket发送给APP 暂时不写

            }

        });
    }

    /**
     * 更新设备影子
     *
     * @param deptId   家庭id/社区园区id
     * @param deviceSn 设备号
     * @param hKey     hKey
     * @param value    value
     */
    private void redisInfo(Long deptId, String deviceSn, String hKey, String value) {
        String ancestor = baseService.findAncestorByDeptId(deptId);
        String key = ancestor.concat(":").concat(deviceSn);
        redisService.setCacheMapValue(key, hKey, value);
    }
}
